import asyncio
import concurrent.futures
import inspect
import json
import logging
import re
import time
import traceback
from collections import deque
from logging.handlers import QueueHandler, QueueListener
from queue import Queue
from threading import Lock
from typing import Any, Callable, Dict, List, Pattern, Tuple, Union

import colorlog
import websockets

from .cmd import Cmd
from .messenger import Messenger
from .msg import Msg


class QPSController:
    """控制最大每秒请求数（QPS）"""

    def __init__(
        self,
        max_qps: int = 3
    ):
        """初始化 QPS 控制实例

        Args:
            max_qps (int, optional): 最大每秒请求数。默认为 3。
        """
        self.max_qps = max_qps
        self.timestamps = deque()

    def can_send(self) -> bool:
        """检查当前是否可发送消息（是否达到 QPS 上限）

        Returns:
            bool: 如果允许发送，返回 True；否则返回 False。
        """
        now = time.time()
        while self.timestamps and now - self.timestamps[0] > 1.0:
            self.timestamps.popleft()
        if len(self.timestamps) < self.max_qps:
            self.timestamps.append(now)
            return True
        return False


class Plugin:
    """插件主类"""

    def __init__(
        self,
        ws_url: str,
        pid: str,
        name: str,
        token: str,
        max_workers: int = 4,
        max_qps: int = 3
    ):
        """初始化 Plugin 实例

        Args:
            ws_url (str): WebSocket 连接地址
            pid (str): 插件唯一标识符
            name (str): 插件名称
            token (str): 插件身份令牌
            max_workers (int, optional): 最大并发线程数量。默认为 4。
            max_qps (int, optional): 一秒内最大消息包发送数量。默认为 3。用于防止发送用户刷屏机器人跟刷的情况。
        """
        self.running = False

        self.seq = 1
        self.seq_lock = Lock()
        self.ws_url = ws_url
        self.pid = pid
        self.name = name
        self.token = token

        self.ws = None
        self.max_workers = max_workers
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)

        self.handlers = {
            "onMsg": [],
            "onCmd": [],
            "onHeartbeat": [],
            "onGroupClockin": [],
            "onBeatABeat": [],
            "onWithdraw": []
        }
        self.handlers_lock = Lock()

        self.response_handlers = {}
        self.response_handlers_lock = Lock()

        self._setup_logging()

        self.last_heartbeat = int(time.time())
        self.total_heartbeat = 0

        self.totalSendNum = {}
        self.sendNum = {}
        self.totalRecvNum = {}
        self.recvNum = {}

        self.golineModes = {}
        self.qps_controller = QPSController(max_qps=3)

    def _setup_logging(self):
        """初始化日志系统，使用 QueueHandler 方式进行异步日志输出

        Raises:
            Exception: 初始化过程中可能抛出的异常
        """
        self.log_queue = Queue(-1)
        self.queue_handler = QueueHandler(self.log_queue)
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.DEBUG)

        for handler in list(self.logger.handlers):
            self.logger.removeHandler(handler)

        self.logger.addHandler(self.queue_handler)

        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.DEBUG)
        console_formatter = colorlog.ColoredFormatter(
            '%(log_color)s[%(asctime)s.%(msecs)03d] %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S',
            log_colors={
                'DEBUG': 'white',
                'INFO': 'green',
                'WARNING': 'yellow',
                'ERROR': 'red',
                'CRITICAL': 'bold_red',
            }
        )
        console_handler.setFormatter(console_formatter)

        file_handler = logging.FileHandler("app.log", encoding="utf-8")
        file_handler.setLevel(logging.DEBUG)
        file_formatter = logging.Formatter(
            '[%(asctime)s.%(msecs)03d] %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        )
        file_handler.setFormatter(file_formatter)

        self.queue_listener = QueueListener(self.log_queue, console_handler, file_handler)
        self.queue_listener.start()

    def _format_exception(
        self,
        e: Exception
    ) -> str:
        """将异常转化为 traceback 字符串

        Args:
            e (Exception): 异常对象

        Returns:
            str: 格式化后的 traceback 字符串

        Raises:
            TypeError: 当输入不是 Exception 实例时
        """
        if isinstance(e, Exception):
            return ''.join(traceback.format_exception(type(e), e, e.__traceback__)).strip()
        return str(e)

    def log(self,
            msg: Any,
            level: int = logging.INFO,
            main_tag: str = "SecPlugin",
            tag: str = None
    ):
        """通用日志记录方法

        Args:
            msg (Any): 待记录的消息内容
            level (int, optional): 日志级别。默认为 INFO。
            main_tag (str, optional): 主标签名。默认为 "SecPlugin"。
            tag (str, optional): 消息标签。如果未传递，则根据日志级别自动生成。
        """
        if tag is None:
            level_name = logging.getLevelName(level)
            tag = f"on{level_name.capitalize()}Message"

        if msg is None:
            msg = "null"
        elif isinstance(msg, dict) or isinstance(msg, list):
            msg = json.dumps(msg, ensure_ascii=False)
        elif isinstance(msg, Messenger):
            msg = json.dumps(msg.getList(), ensure_ascii=False)
        elif isinstance(msg, Exception):
            msg = self._format_exception(msg)
        elif not isinstance(msg, str):
            msg = str(msg)

        formatted_msg = f"[{main_tag}::{tag}] {msg}"
        self.logger.log(level, formatted_msg)

    def getSeq(
        self,
        next: bool = True
    ) -> int:
        """生成或返回当前消息序号（seq）

        Args:
            next (bool, optional): 是否生成下一个序号。默认为 True。

        Returns:
            int: 当前或下一个序号
        """
        with self.seq_lock:
            if next:
                self.seq += 1
            return self.seq

    async def _run(self) -> None:
        """运行核心异步任务，包括连接和消息处理

        Raises:
            asyncio.CancelledError: 当任务被取消时
            Exception: 连接过程中可能抛出的异常
        """
        try:
            self.log(f"正在连接 {self.ws_url}", tag="onConnectWebSocket")
            await self.onConnectWebSocket()
        except asyncio.CancelledError:
            self.log("断开连接", tag="onDisconnectWebSocket")
        finally:
            if self.ws:
                await self.ws.close()
                self.log("关闭连接", tag="onDisconnectWebSocket")

    def start(self) -> None:
        """启动插件主逻辑

        Raises:
            Exception: 启动过程中可能抛出的异常
        """
        try:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            task = loop.create_task(self._run())

            try:
                loop.run_until_complete(task)
            except KeyboardInterrupt:
                task.cancel()
                try:
                    loop.run_until_complete(task)
                except asyncio.CancelledError:
                    pass
            finally:
                self.onClose(loop)
        except Exception as e:
            self.log(e, tag="onStartError", level=logging.ERROR)
        finally:
            if self.running:
                self.onClose(loop)

    async def onConnectWebSocket(self) -> None:
        """建立 WebSocket 连接并开始接收和处理消息

        Raises:
            websockets.exceptions.ConnectionClosed: 连接关闭异常
            json.JSONDecodeError: JSON 解析错误
            Exception: 连接过程中的其他异常
        """
        last_msg = ""
        try:
            async with websockets.connect(self.ws_url) as websocket:
                self.ws = websocket
                self.onOpen()
                self.log("连接成功", tag="onConnectWebSocket")

                async for msg in websocket:
                    try:
                        last_msg = msg;
                        message = json.loads(msg)
                    except json.JSONDecodeError as e:
                        self.log(e, level=logging.WARNING)
                    asyncio.create_task(self.onRecvWssHandler(message))
        except Exception as e:
            self.log(e, tag="onWebSocketError", level=logging.ERROR)

    def onOpen(self) -> None:
        """连接建立后发送上线请求

        Raises:
            Exception: 发送上线请求过程中可能抛出的异常
        """
        try:
            self.running = True
            self.sendWss(Cmd.SyncOicq,
                         {"pid": self.pid, "name": self.name, "token": self.token},
                         rsp=False)
        except Exception as e:
            self.log(e, tag="onOpen", level=logging.ERROR)

    def onClose(
        self,
        loop = None
    ) -> None:
        """关闭插件和相关资源

        Args:
            loop: 可选的事件循环对象，用于关���异步任务

        Raises:
            Exception: 关闭过程中可能抛出的异常
        """
        if loop is not None:
            try:
                for t in asyncio.all_tasks(loop):
                    t.cancel()
                loop.run_until_complete(asyncio.gather(*asyncio.all_tasks(loop), return_exceptions=True))
            except (KeyboardInterrupt, SystemExit):
                pass
            except Exception as e:
                self.log("关闭任务时出错", tag="onClose", level=logging.WARNING)
                self.log(e, tag="onClose", level=logging.WARNING)
            finally:
                try:
                    if not loop.is_closed():
                        loop.close()
                except (KeyboardInterrupt, SystemExit):
                    pass
                except Exception as e:
                    self.log("关闭事件循环时出错", tag="onClose", level=logging.WARNING)
                    self.log(e, tag="onClose", level=logging.WARNING)

        self.executor.shutdown(wait=False)

        if hasattr(self, "queue_listener"):
            self.queue_listener.stop()

        self.running = False
        self.log("插件已关闭", tag="onClose")

    def sendWss(self,
                cmd: str,
                data: Union['Messenger', dict]|None=None,
                rsp: bool=True,
                timeout: int=3
    ):
        """发送websocket消息
        
        args:
            cmd: 命令类型
            data: 消息数据
            rsp: 是否等待响应，对于pushoicqmsg命令默认为false
            timeout: 超时时间（秒）
        """
        if not self.running:
            return
        if not cmd:
            return

        seq = self.getSeq()
        msg = {
            "seq": seq,
            "cmd": cmd,
            "rsp": rsp
        }
        if data:
            if isinstance(data, dict):
                msg["data"] = data
            elif data.getList():
                msg["data"] = data.getList()

        message = json.dumps(msg)
        if data and cmd == Cmd.SendOicqMsg and isinstance(data, Messenger):
            self.totalSendNum[data.get(Msg.Account)] = self.totalSendNum.get(data.get(Msg.Account), 0) + 1
            self.sendNum[data.get(Msg.Account)] = self.sendNum.get(data.get(Msg.Account), {})
            self.sendNum[data.get(Msg.Account)][data.getChatId()] = self.sendNum[data.get(Msg.Account)].get(data.getChatId(), 0) + 1
        try:
            loop = asyncio.get_running_loop()
            loop.create_task(self.ws.send(message))
        except RuntimeError:
            self.executor.submit(lambda: asyncio.run(self.ws.send(message)))
        except Exception as e:
            self.log(e, tag="onSendWss", level=logging.ERROR)
        if rsp:
            future = concurrent.futures.Future()
            with self.response_handlers_lock:
                self.response_handlers[seq] = future
            try:
                result = future.result(timeout=timeout)
                return result
            except concurrent.futures.TimeoutError:
                self.log(f"响应超时 (seq: {seq})", tag="onSendWss", level=logging.WARNING)
            except Exception as e:
                self.log(e, tag="onSendWss", level=logging.ERROR)
            finally:
                with self.response_handlers_lock:
                    if seq in self.response_handlers:
                        del self.response_handlers[seq]

    def sendHeartbeat(self):
        """发送心跳包

        Raises:
            Exception: 发送心跳过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            self.sendWss("Heartbeat", rsp=False)
        except Exception as e:
            self.log(e, tag="sendHeartbeat", level=logging.ERROR)

    def setGroupMemberNick(self,
                           messenger_or_qun: Union['Messenger', str],
                           uin: str,
                           nick: str,
                           account: str = None) -> None:
        """修改群成员昵称

        Args:
            messenger_or_qun (Union[Messenger, str]): 消息对象或群号
            uin (str): 要修改的群成员 QQ 号
            nick (str): 新昵称
            account (str, optional): 账号标识符。默认为 None。

        Returns:
            None: 无返回值

        Raises:
            Exception: 修改昵称过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            if isinstance(messenger_or_qun, Messenger):
                reply = Messenger.getBaseMessenger(messenger_or_qun)
            else:
                reply = Messenger()
                reply.add(Msg.Account, account)
                reply.add(Msg.Group)
                reply.add(Msg.GroupId, str(messenger_or_qun))
            reply.add(Msg.GroupMemberNickModify) \
                 .add(Msg.Uin, uin) \
                 .add(Msg.Nick, nick)
            self.sendWss(Cmd.SendOicqMsg, reply, rsp=False)
        except Exception as e:
            self.log(e, tag="onSendJson", level=logging.ERROR)
        return None

    def withdraw(self,
                 messenger_or_qun: Union['Messenger', str],
                 msgId: str,
                 account: str = None) -> Any:
        """撤回消息

        Args:
            messenger_or_qun (Union[Messenger, str]): 消息对象或群号
            msgId (str): 消息 ID
            account (str, optional): 账号标识符。默认为 None。

        Returns:
            Any: 发送结果或 None

        Raises:
            Exception: 撤回消息过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            if isinstance(messenger_or_qun, Messenger):
                reply = Messenger.getBaseMessenger(messenger_or_qun)
            else:
                reply = Messenger()
                reply.add(Msg.Account, account)
                reply.add(Msg.Group)
                reply.add(Msg.GroupId, str(messenger_or_qun))
            reply.add(Msg.Withdraw, msgId)
            self.sendWss(Cmd.SendOicqMsg, reply, rsp=False)
        except Exception as e:
            self.log(e, tag="onSendJson", level=logging.ERROR)
        return None

    def sendCard(self,
                 messenger: 'Messenger',
                 type: str,
                 *args) -> Any:
        """发送卡片消息

        Args:
            messenger (Messenger): 消息对象
            type (str): 卡片类型
            *args: 卡片内容参数

        Returns:
            Any: 发送结果或 None

        Raises:
            Exception: 发送卡片消息过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            reply = Messenger.getBaseMessenger(messenger)
            reply.add(Msg.CustomJson)
            reply.add(type)
            arguments = [Msg.Title, Msg.Info, Msg.Img, Msg.Url, Msg.Audio]
            match type:
                case Msg.JSON_QQ:
                    for i in range(min(len(args), 5)):
                        reply.add(arguments[i], args[i])
            return self.sendWss(Cmd.SendOicqMsg, reply)
        except Exception as e:
            self.log(e, tag="onSendMsg", level=logging.ERROR)
        return None

    def isOperator(self,
                   messenger_or_qun: Union['Messenger', str],
                   uin: str = None,
                   account: str = None) -> bool:
        """检查是否为管理员

        Args:
            messenger_or_qun (Union[Messenger, str]): 消息对象或群号
            uin (str, optional): QQ 号
            account (str, optional): 账号标识符。默认为 None。

        Returns:
            bool: 是否为管理员

        Raises:
            Exception: 查询管理员列表过程中可能抛出的异常
        """
        if not self.running:
            return False
        try:
            if isinstance(messenger_or_qun, Messenger):
                reply = Messenger.getBaseMessenger(messenger_or_qun)
                if uin is None:
                    uin = messenger_or_qun.get(Msg.Uin)
            else:
                reply = Messenger()
                reply.add(Msg.Account, account)
                reply.add(Msg.Group)
                reply.add(Msg.GroupId, str(messenger_or_qun))
            reply.add(Msg.GroupMemberListGetAdmin)
            operators = self.sendWss(Cmd.SendOicqMsg, reply, rsp=True)
            if operators is None:
                return False
            operators = operators.get("data", [])
            for operator in operators:
                if operator.get(Msg.Uin, "") == uin:
                    return True
        except Exception as e:
            self.log(e, tag="getIsOperator", level=logging.ERROR)
        return False

    def sendMsg(self,
                messenger: 'Messenger',
                *text: str,
                replyMsgId: int = 0) -> Any:
        """发送文本消息（支持插入图片）

        Args:
            messenger (Messenger): 消息对象
            *text (str): 消息文本（多个参数）
            replyMsgId (int, optional): 回复的消息 ID。默认为 0。

        Returns:
            Any: 发送结果或 None

        Raises:
            Exception: 发送消息过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            pattern = "\\[图片=(.+)\\]"
            result = []
            for t in text:
                if not t:
                    continue
                if isinstance(t, (dict, list)):
                    t = json.dumps(t, ensure_ascii=False)
                elif isinstance(t, Messenger):
                    t = json.dumps(t.getList(), ensure_ascii=False)
                elif isinstance(t, Exception):
                    t = self._format_exception(t)
                elif not isinstance(t, str):
                    t = str(t)
                matches = list(re.finditer(pattern, t))
                if matches:
                    last_end = 0
                    for match in matches:
                        start, end = match.span()
                        result.append(t[last_end:start])
                        result.append(f"[图片={match.group(1)}]")
                        last_end = end
                    result.append(t[last_end:])
                else:
                    result.append(t)
            reply = Messenger.getBaseMessenger(messenger)
            for item in result:
                if item.startswith("[图片=") and item.endswith("]"):
                    link = item[4:-1]
                    reply.add(Msg.Img, link)
                else:
                    reply.add(Msg.Text, item)
            if replyMsgId:
                reply.add(Msg.Reply, replyMsgId)
            return self.sendWss(Cmd.SendOicqMsg, reply)
        except Exception as e:
            self.log(e, tag="onSendMsg", level=logging.ERROR)
        return None

    def sendReplyMsg(self,
                     messenger: 'Messenger',
                     *text: str,
                     replyMsgId: int = 0) -> Any:
        """发送回复消息

        Args:
            messenger (Messenger): 消息对象
            *text (str): 消息文本（多个参数）
            replyMsgId (int, optional): 明确指定的消息 ID。若未指定，则默认回复当前消息。

        Returns:
            Any: 发送结果或 None

        Raises:
            Exception: 发送消息过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            if not replyMsgId:
                replyMsgId = messenger.get(Msg.MsgId)
            return self.sendMsg(messenger, *text, replyMsgId=replyMsgId)
        except Exception as e:
            self.log(e, tag="onSendMsg", level=logging.ERROR)
        return None

    def sendJson(self,
                 messenger: 'Messenger',
                 *json_text: str) -> List[Any]:
        """发送 JSON 卡片消息

        Args:
            messenger (Messenger): 消息对象
            *json_text (str): JSON 消息内容

        Returns:
            List[Any]: 发送结果列表

        Raises:
            Exception: 发送 JSON 消息过程中可能抛出的异常
        """
        if not self.running:
            return []
        try:
            results = []
            for jsont in json_text:
                reply = Messenger.getBaseMessenger(messenger) \
                                .add(Msg.Json, jsont)
                results.append(self.sendWss(Cmd.SendOicqMsg, reply))
            return results
        except Exception as e:
            self.log(e, tag="onSendJson", level=logging.ERROR)
        return []

    def sendImg(self,
                messenger: 'Messenger',
                *url: str) -> Any:
        """发送图片消息

        Args:
            messenger (Messenger): 消息对象
            *url (str): 图片链接列表

        Returns:
            Any: 发送结果或 None

        Raises:
            Exception: 发送图片消息过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            for u in url:
                reply = Messenger.getBaseMessenger(messenger) \
                                .add(Msg.Img, u)
            return self.sendWss(Cmd.SendOicqMsg, reply)
        except Exception as e:
            self.log(e, tag="onSendImg", level=logging.ERROR)
        return None

    def getTotalSend(self,
                     messenger: 'Messenger') -> int:
        """获取总发送数量

        Args:
            messenger (Messenger): 消息对象

        Returns:
            int: 总发送量
        """
        return self.totalSendNum.get(messenger.get(Msg.Account), 0)

    def getSend(self,
                messenger: 'Messenger') -> int:
        """获取指定会话的发送数量

        Args:
            messenger (Messenger): 消息对象

        Returns:
            int: 该会话的发送量
        """
        account = messenger.get(Msg.Account)
        chat_id = messenger.getChatId()
        return self.sendNum.get(account, {}).get(chat_id, 0)

    def getTotalRecv(self,
                     messenger: 'Messenger') -> int:
        """获取总接收数量

        Args:
            messenger (Messenger): 消息对象

        Returns:
            int: 总接收量
        """
        return self.totalRecvNum.get(messenger.get(Msg.Account), 0)

    def getRecv(self,
                messenger: 'Messenger') -> int:
        """获取指定会话的接收数量

        Args:
            messenger (Messenger): 消息对象

        Returns:
            int: 该会话的接收量
        """
        account = messenger.get(Msg.Account)
        chat_id = messenger.getChatId()
        return self.recvNum.get(account, {}).get(chat_id, 0)

    def onMsg(self, msg: Union[str, Pattern]):
        """注册消息处理函数（正则匹配）

        Args:
            msg (Union[str, Pattern]): 匹配规则（字符串或编译好的Pattern对象）

        Returns:
            function: 装饰器函数
        """
        compiled_pattern = re.compile(msg)
        def decorator(func):
            with self.handlers_lock:
                self.handlers["onMsg"].append((
                    compiled_pattern,
                    func
                ))
            return func
        return decorator

    def onCmd(self, cmd: str):
        """��册命令处理函数（精确命令匹配）

        Args:
            cmd (str): 消息命令类型

        Returns:
            function: 装饰器函数
        """
        def decorator(func):
            with self.handlers_lock:
                self.handlers["onCmd"].append((
                    cmd,
                    func
                ))
            return func
        return decorator

    def onHeartbeat(self):
        """注册心跳处理函数

        Returns:
            function: 装饰器函数
        """
        def decorator(func):
            with self.handlers_lock:
                self.handlers["onHeartbeat"].append(func)
            return func
        return decorator

    def onGroupClockin(self):
        """注册群打卡处理函数

        Returns:
            function: 装饰器函数
        """
        def decorator(func):
            with self.handlers_lock:
                self.handlers["onGroupClockin"].append(func)
            return func
        return decorator

    def onBeatABeat(self):
        """注册戳一戳处理函数

        Returns:
            function: 装饰器函数
        """
        def decorator(func):
            with self.handlers_lock:
                self.handlers["onBeatABeat"].append(func)
            return func
        return decorator

    def onWithdraw(self):
        """注册消息撤回处理函数

        Returns:
            function: 装饰器函数
        """
        def decorator(func):
            with self.handlers_lock:
                self.handlers["onWithdraw"].append(func)
            return func
        return decorator

    async def onRecvWssHandler(self, msg: Dict[str, Any]) -> None:
        """处理接收到的 WebSocket 消息主线程，按类型分发不同处理器

        Args:
            msg (Dict[str, Any]): 收到的消息 JSON 数据

        Raises:
            Exception: 消息处理过程中可能抛出的异常
        """
        if not self.running:
            return
        cmd = msg.get("cmd")
        await self.onCmdHandler(msg)
        if cmd == Cmd.PushOicqMsg:
            messenger = Messenger(msg.get("data", []))
            if messenger.has(Msg.System):
                await self.onSystemMsgHandler(messenger)
            else:
                await self.onMsgHandler(messenger)

    async def onCmdHandler(self, msg: Dict[str, Any]) -> None:
        """处理收到的通用命令消息

        Args:
            msg (Dict[str, Any]): 收到的消息 JSON 数据

        Raises:
            Exception: 命令处理过程中可能抛出的异常
        """
        if not self.running:
            return
        cmd = msg.get("cmd")
        match cmd:
            case Cmd.Response:
                await self.onRespMsgHandler(msg)
            case Cmd.Heartbeat:
                await self.onHeartbeatHandler(msg)
            case Cmd.PushOicqMsg:
                pass  # 该类消息已被分配在 onSystemMsgHandler 和 onMsgHandler 内处理
        await self.doCmdHandler(msg)

    async def doCmdHandler(self, msg: Dict[str, Any]) -> None:
        """执行 onCmd 注册的处理函数

        Args:
            msg (Dict[str, Any]): 收到的消息 JSON 数据

        Raises:
            Exception: 处理函数执行过程中可能抛出的异常
        """
        if not self.running:
            return
        start_time = int(round(time.time() * 1000))
        try:
            for cmd, callback in self.handlers["onCmd"]:
                if cmd == msg.get("cmd"):
                    loop = asyncio.get_running_loop()
                    if inspect.iscoroutinefunction(callback):
                        await callback(msg)
                    else:
                        await loop.run_in_executor(self.executor, callback, msg)
                    self.log(f"消息处理耗时 {(int(round(time.time() * 1000)) - start_time) / 1000} 秒", tag="doCmdHandler")
        except Exception as e:
            self.log(e, tag="doCmdHandler", level=logging.ERROR)

    async def onRespMsgHandler(self, msg: Dict[str, Any]) -> None:
        """处理响应消息并唤醒等待的 Future

        Args:
            msg (Dict[str, Any]): 收到的响应消息 JSON 数据

        Raises:
            Exception: 响应处理过程中可能抛出的异常
        """
        if not self.running:
            return
        if isinstance(msg.get("data"), dict):
            if "status" in msg.get("data"):
                if msg.get("data").get("status"):
                    self.log("对接成功", tag="onSyncOicq")
                else:
                    self.log("口令错误", tag="onSyncOicq", level=logging.ERROR)

        seq = msg.get("seq")
        if seq:
            with self.response_handlers_lock:
                if seq in self.response_handlers:
                    future = self.response_handlers[seq]
                    future.set_result(msg)
                    del self.response_handlers[seq]

    async def onSystemMsgHandler(self, messenger: 'Messenger') -> None:
        """处理系统消息（如上线、下线、心跳等）

        Args:
            messenger (Messenger): 消息对象

        Raises:
            Exception: 系统消息处理过程中可能抛出的异常
        """
        if not self.running:
            return
        if messenger.has(Msg.Goline) or messenger.has(Msg.Offline) or messenger.has(Msg.Heartbeat) or messenger.has(Msg.OntimeTask):
            await self.onAccountEvent(messenger)
        else:
            await self.onUnsupportedMsg(messenger)

    async def onAccountEvent(self, messenger: 'Messenger') -> None:
        """处理账号事件，如上线、下线、心跳等

        Args:
            messenger (Messenger): 消息对象

        Raises:
            Exception: 账号事件处理过��中可能抛出的异常
        """
        if not self.running:
            return
        log_msg = f"{messenger.get(Msg.Account)} "
        non_set = log_msg
        if messenger.has(Msg.Goline):
            log_msg += f"账号状态变更为上线({messenger.get(Msg.GolineMode)})"
        elif messenger.has(Msg.Offline):
            log_msg += "账号状态变更为下线"
        elif messenger.has(Msg.Heartbeat):
            log_msg += "账号发生心跳"
        elif messenger.has(Msg.OntimeTask):
            log_msg += "账号触发定时任务"
        if log_msg != non_set:
            self.log(log_msg, tag="onAccountEvent", level=logging.DEBUG)

    async def onUnsupportedMsg(self, messenger) -> None:
        """处理未知消息类型并记录警告日志

        Args:
            messenger: 消息对象

        Raises:
            Exception: 日志记录过程中可能抛出的异常
        """
        self.log(messenger, tag="onUnsupportedMsg", level=logging.WARNING)

    def getFunctionRequiredNum(self, callback: Callable) -> int:
        """提取函数所需参数数量，用于适配参数传递逻辑

        Args:
            callback (Callable): 回调函数

        Returns:
            int: 所需参数数量（不包含默认值参数）

        Raises:
            TypeError: 若函数包含 *args 或 **kwargs，抛出异常
        """
        sig = inspect.signature(callback)
        if any(p.kind in (p.VAR_POSITIONAL, p.VAR_KEYWORD) for p in sig.parameters.values()):
            raise TypeError(f"{callback.__name__} 不能使用 *args 或 **kwargs")
        return sum(
            1 for param in sig.parameters.values()
            if param.default == inspect.Parameter.empty
            and param.kind in (param.POSITIONAL_ONLY, param.POSITIONAL_OR_KEYWORD)
        )

    async def runFunction(self, callback: Callable, *args) -> Any:
        """根据函数类型（异步/同步）执行处理函数

        Args:
            callback (Callable): 处理函数
            *args: 参数

        Returns:
            Any: 函数返回值

        Raises:
            TypeError: 参数数量不匹配时抛出异常
            Exception: 函数执行过程中可能抛出的异常
        """
        required_count = self.getFunctionRequiredNum(callback)
        if len(args) != required_count:
            raise TypeError(f"{callback.__name__} 需要{required_count}参数，传递{len(args)}参数")
        loop = asyncio.get_running_loop()
        returns = None
        if inspect.iscoroutinefunction(callback):
            returns = await callback(*args)
        else:
            returns = await loop.run_in_executor(self.executor, callback, *args)
        return returns

    async def onMsgHandler(self, messenger: 'Messenger') -> None:
        """处理文本群消息并调用对应处理函数

        Args:
            messenger (Messenger): 消息对象

        Raises:
            Exception: 消息处理过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            text = ""
            urlList = messenger.getList(Msg.Url)
            emoyList = messenger.getList(Msg.Emoy)
            imgI = gifI = emoyI = 0
            for dat in messenger.getList():
                if Msg.Text in dat:
                    text += dat.get(Msg.Text)
                if Msg.Img in dat:
                    text += f"[{'动图' if dat.get(Msg.Img).endswith('.gif') else '图片'}={urlList[imgI]},尺寸={messenger.getList(Msg.Width)[imgI]}x{messenger.getList(Msg.Height)[imgI]},大小={messenger.getList(Msg.Size)[imgI]}B]"
                    imgI += 1
                if Msg.Gif in dat:
                    text += f"[动图={urlList[gifI]},尺寸={messenger.getList(Msg.Width)[gifI]}x{messenger.getList(Msg.Height)[gifI]},大小={messenger.getList(Msg.Size)[gifI]}B]"
                    gifI += 1
                if Msg.Ptt in dat:
                    text += f"[语音,大小={messenger.get(Msg.Size)}B]"
                if Msg.Video in dat:
                    text += f"[视频,尺寸={messenger.get(Msg.Width)}x{messenger.get(Msg.Height)},大小={messenger.get(Msg.Size)}B]"
                if Msg.Emoy in dat:
                    text += f"[表情,id={emoyList[emoyI]}]"
                    emoyI += 1
            if not text:
                await self.onEventMsg(messenger)
                return
            else:
                self.log(f"{messenger.get(Msg.GroupName)} - {messenger.get(Msg.UinName)}: {text}", tag=f"on{Messenger.getMsgType(messenger).capitalize()}Msg", level=logging.DEBUG)

            self.totalRecvNum[messenger.get(Msg.Account)] = self.totalRecvNum.get(messenger.get(Msg.Account), 0) + 1
            self.recvNum.setdefault(messenger.get(Msg.Account), {})
            self.recvNum[messenger.get(Msg.Account)][messenger.getChatId()] = \
                self.recvNum[messenger.get(Msg.Account)].get(messenger.getChatId(), 0) + 1
            await self.doMsgHandler(messenger)
        except Exception as e:
            self.log(e, tag="onMsgHandler", level=logging.ERROR)

    async def doMsgHandler(self, messenger: 'Messenger') -> None:
        """执行文本消息匹配并运行相应处理函数

        Args:
            messenger (Messenger): 消息对象

        Raises:
            TypeError: 处理函数参数数量不匹配时抛出异常
            Exception: 处理函数执行过程中可能抛出的异常
        """
        if not self.running:
            return
        start_time = int(round(time.time() * 1000))
        text = messenger.get(Msg.Text)
        try:
            if not text:
                return
            for regex, callback in self.handlers["onMsg"]:
                matches = re.fullmatch(regex, text)
                if matches:
                    required_count = self.getFunctionRequiredNum(callback)
                    args = ()
                    if required_count == 1:
                        args = (messenger,)
                    elif required_count == 2:
                        args = (messenger, matches)
                    elif required_count:
                        raise TypeError(f"{callback.__name__} 应当定义为不需要传递实参或只传递1或2个实参的函数")
                    await self.runFunction(callback, *args)
                    self.log(f"消息处理耗时 {(int(round(time.time() * 1000)) - start_time) / 1000} 秒", tag="doMsgHandler")
        except Exception as e:
            self.log(e, tag="doMsgHandler", level=logging.ERROR)

    async def onHeartbeatHandler(self, msg: dict) -> None:
        """处理心跳响应并广播回复心跳包

        Args:
            msg (dict): 心跳消息数据

        Raises:
            Exception: 心跳处理过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            msg_time = int(time.time())
            self.log("接收到心跳包", tag="onHeartbeatHandler")
            self.log(f"距上次接收到 {msg_time - self.last_heartbeat} 秒", tag="onHeartbeatHandler")
            self.last_heartbeat = msg_time
            self.total_heartbeat += 1
            self.log(f"总计心跳次数 {self.total_heartbeat}", tag="onHeartbeatHandler")
            self.sendHeartbeat()

            await self.doHeartbeatHandler(msg)
        except Exception as e:
            self.log(e, tag="onHeartbeatHandler", level=logging.ERROR)

    async def doHeartbeatHandler(self, msg: dict) -> None:
        """执行心跳注册处理函数

        Args:
            msg (dict): 心跳消息数据

        Raises:
            Exception: 心跳处理函数执行过程中可能抛出的异常
        """
        if not self.running:
            return
        start_time = int(round(time.time() * 1000))
        try:
            for callback in self.handlers["onHeartbeat"]:
                await self.runFunction(callback)
                self.log(f"消息处理耗时 {(int(round(time.time() * 1000)) - start_time) / 1000} 秒", tag="doHeartbeatHandler")
        except Exception as e:
            self.log(e, tag="doHeartbeatHandler", level=logging.ERROR)

    async def onEventMsg(self, messenger: 'Messenger') -> None:
        """根据系统事件消息分发并运行相应处理器

        Args:
            messenger (Messenger): 消息对象

        Raises:
            Exception: 事件消息处理过程中可能抛出的异常
        """
        if not self.running:
            return
        if messenger.has(Msg.GroupClockin):
            await self.onGroupClockinHandler(messenger)
        elif messenger.has(Msg.GroupBeatABeat) or messenger.has(Msg.FriendBeatABeat):
            await self.onBeatABeatHandler(messenger)
        elif messenger.has(Msg.Withdraw):
            await self.onWithdrawHandler(messenger)
        else:
            await self.onUnsupportedMsg(messenger)

    async def onGroupClockinHandler(self, messenger: 'Messenger') -> None:
        """处理群打卡事件消息

        Args:
            messenger (Messenger): 消息对象

        Raises:
            Exception: 打卡事件处理过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            self.log(f"{messenger.get(Msg.GroupName)} - {messenger.get(Msg.UinName)}({messenger.get(Msg.Uin)}) {messenger.get(Msg.GroupClockin)}", tag="onGroupClockinHandler", level=logging.DEBUG)

            await self.doGroupClockinHandler(messenger)
        except Exception as e:
            self.log(e, tag="onGroupClockinHandler", level=logging.ERROR)

    async def doGroupClockinHandler(self, messenger: 'Messenger') -> None:
        """执行群打卡注册的处理函数

        Args:
            messenger (Messenger): 消息对象

        Raises:
            TypeError: 处理函数参数数量不匹配时抛出异常
            Exception: 打卡处理函数执行过程中可能抛出的异常
        """
        if not self.running:
            return
        start_time = int(round(time.time() * 1000))
        try:
            for callback in self.handlers["onGroupClockin"]:
                required_count = self.getFunctionRequiredNum(callback)
                args = ()
                if required_count == 1:
                    args = (messenger,)
                elif required_count:
                    raise TypeError(f"{callback.__name__} 应当定义为不需要传递实参或只传递1个实参���函数")
                await self.runFunction(callback, *args)
                self.log(f"消息处理耗时 {(int(round(time.time() * 1000)) - start_time) / 1000} 秒", tag="doGroupClockinHandler")
        except Exception as e:
            self.log(e, tag="doGroupClockinHandler", level=logging.ERROR)

    async def onBeatABeatHandler(self, messenger: 'Messenger') -> None:
        """处理戳一戳事件消息

        Args:
            messenger (Messenger): 消息对象

        Raises:
            Exception: 戳一戳事件处理过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            match Messenger.getMsgType(messenger):
                case Msg.Group:
                    self.log(
                        f"{messenger.get(Msg.GroupName)} - {messenger.get(Msg.OpName)}({messenger.get(Msg.Op)}) 双戳事件：{messenger.get(Msg.GroupBeatABeat)} {messenger.get(Msg.UinName)}({messenger.get(Msg.Uin)})",
                        tag="onGroupBeatABeatHandler",
                        level=logging.DEBUG)
                case Msg.Friend:
                    self.log(
                        f"{messenger.get(Msg.UinName)}({messenger.get(Msg.Uin)}) 戳了戳 bot({messenger.get(Msg.Account)})",
                        tag="onFriendBeatABeatHandler",
                        level=logging.DEBUG)
            await self.doBeatABeatHandler(messenger)
        except Exception as e:
            self.log(e, tag="onBeatABeatHandler", level=logging.ERROR)

    async def doBeatABeatHandler(self, messenger: 'Messenger') -> None:
        """执行双戳事件注册的处理函数

        Args:
            messenger (Messenger): 消息对象

        Raises:
            TypeError: 处理函数参数数量不匹配时抛出异常
            Exception: 双戳处理函数执行过程中可能抛出的异常
        """
        if not self.running:
            return
        start_time = int(round(time.time() * 1000))
        try:
            for callback in self.handlers["onBeatABeat"]:
                required_count = self.getFunctionRequiredNum(callback)
                args = ()
                if required_count == 1:
                    args = (messenger,)
                elif required_count:
                    raise TypeError(f"{callback.__name__} 应当定义为不需要传递实参或只传递1个实参的函数")
                await self.runFunction(callback, *args)
                self.log(f"消息处理耗时 {(int(round(time.time() * 1000)) - start_time) / 1000} 秒", tag="doBeatABeatHandler")
        except Exception as e:
            self.log(e, tag="doBeatABeatHandler", level=logging.ERROR)

    async def onWithdrawHandler(self, messenger: 'Messenger') -> None:
        """处理消息撤回事件

        Args:
            messenger (Messenger): 消息对象

        Raises:
            Exception: 撤回事件处理过程中可能抛出的异常
        """
        if not self.running:
            return
        try:
            text = ""
            if messenger.has(Msg.Group):
                text += f"{messenger.get(Msg.GroupName)} - "
            if messenger.get(Msg.Uin) != messenger.get(Msg.Op):
                text += f"{messenger.get(Msg.OpName)}({messenger.get(Msg.Op)}) 撤回了 {messenger.get(Msg.UinName)}({messenger.get(Msg.Uin)}) 的消息(MsgId={messenger.get(Msg.Withdraw)})"
            else:
                text += f"{messenger.get(Msg.UinName)}({messenger.get(Msg.Uin)}) 撤回了一条消息(MsgId={messenger.get(Msg.Withdraw)})"
            self.log(text, tag="onWithdrawHandler", level=logging.DEBUG)

            await self.doWithdrawHandler(messenger)
        except Exception as e:
            self.log(e, tag="onWithdrawHandler", level=logging.ERROR)

    async def doWithdrawHandler(self, messenger: 'Messenger') -> None:
        """执行消息撤回处理函数

        Args:
            messenger (Messenger): 消息对象

        Raises:
            TypeError: 处理函数参数数量不匹配时抛出异常
            Exception: 撤回处理函数执行过程中可能抛出的异常
        """
        if not self.running:
            return
        start_time = int(round(time.time() * 1000))
        try:
            for callback in self.handlers["onWithdraw"]:
                required_count = self.getFunctionRequiredNum(callback)
                args = ()
                if required_count == 1:
                    args = (messenger,)
                elif required_count:
                    raise TypeError(f"{callback.__name__} 应当定义为不需要传递实参或只传递1个实参的函数")
                await self.runFunction(callback, *args)
                self.log(f"消息处理耗时 {(int(round(time.time() * 1000)) - start_time) / 1000} 秒", tag="doWithdrawHandler")
        except Exception as e:
            self.log(e, tag="doWithdrawHandler", level=logging.ERROR)
